package com.ta.gulimall.coupon.job.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.TypeReference;
import com.ta.gulimall.common.utils.RedisUtil;
import com.ta.gulimall.coupon.config.rabbitmq.MqQueueConstant;
import com.ta.gulimall.coupon.constant.CouponConstants;
import com.ta.gulimall.coupon.feign.ProductFeignService;
import com.ta.gulimall.coupon.job.CouponJobService;
import com.ta.gulimall.coupon.service.SeckillSessionService;
import com.ta.gulimall.coupon.view.vo.SeckillSkuRedisVo;
import com.ta.gulimall.coupon.view.vo.SkuInfoVo;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

@Service
public class CouponJobServiceImpl implements CouponJobService {
    @Autowired
    SeckillSessionService seckillSessionService;
    @Autowired
    ProductFeignService productFeignService;
    @Autowired
    RedisUtil redisUtil;
    @Autowired
    RedissonClient redissonClient;
    @Autowired(required = false)
    RabbitTemplate rabbitTemplate;

    @Override
    @Scheduled(cron = "0 */1 * * * ?")
    public void seckillJobDispatch() {
        //未来秒杀的商品集合
        List<SeckillSkuRedisVo> withSkuses = seckillSessionService.querySeckillGoods();

        if(CollUtil.isEmpty(withSkuses)){
            return;
        }

        //skuId集合
        List<Long> skuIds = withSkuses.stream().map(SeckillSkuRedisVo::getSkuId).distinct().collect(Collectors.toList());

        //sku信息
        List<SkuInfoVo> skuInfoVos = productFeignService.querySpuInfo(skuIds).getData(new TypeReference<List<SkuInfoVo>>() {});
        Map<Long, SkuInfoVo> skuInfoVoMap = skuInfoVos.stream().collect(Collectors.toMap(SkuInfoVo::getSkuId, Function.identity()));

        LocalDateTime now = LocalDateTime.now();
        for (SeckillSkuRedisVo skuRedisVo : withSkuses) {
            Duration duration = Duration.between(now, skuRedisVo.getEndTime());
            SkuInfoVo skuInfoVo = skuInfoVoMap.get(skuRedisVo.getSkuId());
            redisUtil.hset(
                    CouponConstants.REDIS_SECKILL_GOODS,
                    skuRedisVo.getPromotionSessionId() + ":" + skuRedisVo.getSkuId(),
                    skuRedisVo.setSkuInfoVo(skuInfoVo),
                    duration.getSeconds()
            );
            //信号量限流
            RSemaphore semaphore = redissonClient.getSemaphore(CouponConstants.REDIS_SECKILL_SEMAPHORE + ":" + skuRedisVo.getPromotionSessionId() + skuRedisVo.getSkuId());
            // 商品可以秒杀的数量作为信号量
            semaphore.trySetPermits(skuRedisVo.getSeckillCount());
            // 设置信号量的过期时间
            redisUtil.expire(CouponConstants.REDIS_SECKILL_SEMAPHORE + ":" + skuRedisVo.getPromotionSessionId() + skuRedisVo.getSkuId(), duration.getSeconds() + 60);

        }
        // mq延迟消息
        List<SeckillSkuRedisVo> skuRedisVos = withSkuses.stream()
                .collect(Collectors.groupingBy(
                        SeckillSkuRedisVo::getPromotionSessionId,
                        Collectors.collectingAndThen(Collectors.toList(), seckillSkuRedisList -> seckillSkuRedisList.stream().findFirst().orElse(null))
                ))
                .values().stream()
                .sorted((SkuRedisVo1, SkuRedisVo2) -> {
                    LocalDateTime time1 = SkuRedisVo1.getEndTime();
                    LocalDateTime time2 = SkuRedisVo2.getEndTime();
                    return time1.isBefore(time2) ? -1 : time1.isEqual(time2) ? 0 : 1;
                })
                .collect(Collectors.toList());
        //发送消息
        skuRedisVos.forEach(sku -> sendSeckillSession(sku, now));


    }
    void sendSeckillSession(SeckillSkuRedisVo seckillSkuRedisVo, LocalDateTime now){
        String body = CouponConstants.REDIS_SECKILL_SEMAPHORE + ":" + seckillSkuRedisVo.getPromotionSessionId();
        Long millisecond = Duration.between(now, seckillSkuRedisVo.getEndTime()).getSeconds() * 1000L;
        rabbitTemplate.convertAndSend(MqQueueConstant.DEFAULT_DIRECT_EXCHANGE, MqQueueConstant.SECKILL_STORE_DELAYED_QUEUE, body, (message) -> {
            message.getMessageProperties().setExpiration(millisecond.toString());
            return message;
        });
    }
}

